package com.study.kafka;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.Random;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

/**
 * @author yuantongqin
 * 2021/8/4
 **/
public class ProducerMain {

    public static KafkaProducer<String,String> createProducer(){
        Properties properties = new Properties();
        properties.put("bootstrap.servers","172.24.36.103:9092");
        properties.put("acks","1");
        properties.put("retries",3);
//        properties.put("batch.size",)
//        properties.put("linger.ms",)
//        properties.put("client.id",)
//        properties.put("buffer.memory",)
//        producerConfig 是生产者的一些配置文件信息
//        ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG
        properties.put("key.serializer", StringSerializer.class.getName());
        properties.put("value.serializer",StringSerializer.class.getName());

        // 创建kafkaProducer对象
        // 因为我们的消息key和value都是String类型，，所以创建的 Producer 是 <String, String> 的泛型。
        return new KafkaProducer<>(properties);
    }

    public static void main(String[] args) throws ExecutionException, InterruptedException {

        // 创建kafka生成者
        KafkaProducer<String, String> producer = createProducer();
        // 创建消息，传入的三个参数，分别是topic，消息key,消息体
        String topic = "TestTopic";
        Random random = new Random();
        String key = "key"+ random.nextInt(10);
        String value = "kafka 学习"+random.nextInt(100);
        ProducerRecord<String, String> producerRecord =
                new ProducerRecord<String, String>(topic,key,value);

        Future<RecordMetadata> future = producer.send(producerRecord, new Callback() {
            @Override
            public void onCompletion(RecordMetadata recordMetadata, Exception e) {

                System.out.println("回调函数：" + recordMetadata.offset());
            }
        });
        RecordMetadata recordMetadata = future.get();
        StringBuilder sb = new StringBuilder("同步发送的最终结果");
        sb.append("topic:").append(recordMetadata.topic())
                .append(" partition:").append(recordMetadata.partition())
                .append(" offset:").append(recordMetadata.offset());
        System.out.println(sb.toString());

    }
}
